package org.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/*
 Spark RDD的创建
  RDD会贯穿整个Spark学习过程，RDD是Spark最基本存储单元，
  RDD:弹性分布式存储集
 */
object CreateRDD01 {



  def main(args: Array[String]): Unit = {
    //1.准备环境
    val sparkConf: SparkConf = new SparkConf().
      setMaster("local[*]").setAppName("Spark")
    //设置分区数
    //sparkConf.set("spark.default.parallelism","4")
    //设置运行内存
    //sparkConf.set("spark.testing.memory","471859200")
    //2.让SparkContext去加载配置
    val sparkContext: SparkContext = new SparkContext(sparkConf)
    sparkContext.setLogLevel("ERROR")//设置日志级别
    //3.1从内存中创建RDD
    //memoryCreateRDD(sparkContext)
    //3.2从外部(文件)创建RDD
    fileCreateRDD(sparkContext);

    //4.最后一步 关闭环境
    sparkContext.stop()
  }
  def memoryCreateRDD(sparkContext: SparkContext): Unit = {
      //创建RDD  --- 从内存中创建RDD,将内存中集合的数据作为处理的数据源
      val rdd1: RDD[Int] = sparkContext.parallelize(1 to 10)//采用默认分区创建RDD
      println(rdd1.getNumPartitions) //获得Spark RDD底层分区数
      val rdd2: RDD[Int] = sparkContext.parallelize(1 to 10, 3)//3:分区数。在创建RDD的时候指定分区数
      println(rdd2.partitions.length)//获得Spark RDD底层分区数
      //将计算的结果保存到文件当中
      //rdd1.saveAsTextFile("output")
      //makrRDD 和 parallelize 是一样的，parallelize是makeRDD的底层写好
      //将Scala中的数组或者集合 转换成一个RDD对象
      val rdd3: RDD[Int] = sparkContext.makeRDD(1 to 10)

      val arrs: Array[Int] = rdd3.collect()
       arrs.foreach( println )
      //    数组   |   对数组进行遍历
      rdd3.collect().foreach( println )
  }

  def fileCreateRDD(sparkContext: SparkContext) = {
    //1.在项目目录上创建input目录，并在该目录中创建1.txt 在里面写一些东西
    val rdd4: RDD[String] = sparkContext.textFile("input/1.txt")

    rdd4.collect().foreach(println)

  }

}
